/*-------------------------------------------------------------------------
 *
 * basebackup_copy.c
 *	  使用 COPY OUT 发送基备份归档
 *
 * 在开始 COPY OUT 之前，我们发送一个结果集，其中包含将包含
 * 在备份中的表空间信息。然后，我们开始一个单一的 COPY OUT
 * 操作，并在该单一 COPY OUT 过程中传输所有归档和清单（如果存在）。
 * 每个 CopyData 消息以一个类型字节开头，允许我们通过某种方式
 * 信号开始新的归档或清单，而不是通过结束 COPY 流。这也允许
 * 将来的协议扩展，因为只要我们确定客户端会知道该如何处理，
 * 我们可以在消息流中包含任意信息。
 *
 * 不再支持以前使用单独 COPY OUT 操作发送每个归档的旧方法。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/backup/basebackup_copy.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "backup/basebackup.h"
#include "backup/basebackup_sink.h"
#include "catalog/pg_type_d.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "utils/timestamp.h"

typedef struct bbsink_copystream
{
	/* 所有类型的接收器的通用信息。 */
	bbsink		base;

	/* 我们是在将档案发送给客户端，还是发送到其他地方？ */
	bool		send_to_client;

	/*
	 * 协议消息缓冲区。我们通过将此缓冲区的第一个字符设置为'd'（档案或清单数据）来组装CopyData协议消息，然后让base.bbs_buffer指向第二个字符，以便其余数据正好复制到我们希望的位置。
	 */
	char	   *msgbuffer;

	/*
	 * 我们上一次向客户端报告进度是什么时候，报告了多少进度？
	 */
	TimestampTz last_progress_report_time;
	uint64		bytes_done_at_last_time_check;
} bbsink_copystream;

/*
 * 我们不想过于频繁地向客户端发送进度消息。理想情况下，我们希望在自上次消息以来的时间达到PROGRESS_REPORT_MILLISECOND_THRESHOLD时发送消息，但是每次发送一点数据时检查系统时间似乎成本太高。因此，我们只在自上次检查以来的字节数达到PROGRESS_REPORT_BYTE_INTERVAL后检查它。
 */
#define	PROGRESS_REPORT_BYTE_INTERVAL				65536
#define PROGRESS_REPORT_MILLISECOND_THRESHOLD		1000

static void fc_bbsink_copystream_begin_backup(bbsink *fc_sink);
static void fc_bbsink_copystream_begin_archive(bbsink *fc_sink,
											const char *fc_archive_name);
static void fc_bbsink_copystream_archive_contents(bbsink *fc_sink, size_t fc_len);
static void fc_bbsink_copystream_end_archive(bbsink *fc_sink);
static void fc_bbsink_copystream_begin_manifest(bbsink *fc_sink);
static void fc_bbsink_copystream_manifest_contents(bbsink *fc_sink, size_t fc_len);
static void fc_bbsink_copystream_end_manifest(bbsink *fc_sink);
static void fc_bbsink_copystream_end_backup(bbsink *fc_sink, XLogRecPtr fc_endptr,
										 TimeLineID fc_endtli);
static void fc_bbsink_copystream_cleanup(bbsink *fc_sink);

static void fc_SendCopyOutResponse(void);
static void fc_SendCopyDone(void);
static void fc_SendXlogRecPtrResult(XLogRecPtr fc_ptr, TimeLineID fc_tli);
static void fc_SendTablespaceList(List *fc_tablespaces);
static void fc_send_int8_string(StringInfoData *fc_buf, int64 fc_intval);

static const bbsink_ops bbsink_copystream_ops = {
	.begin_backup = fc_bbsink_copystream_begin_backup,
	.begin_archive = fc_bbsink_copystream_begin_archive,
	.archive_contents = fc_bbsink_copystream_archive_contents,
	.end_archive = fc_bbsink_copystream_end_archive,
	.begin_manifest = fc_bbsink_copystream_begin_manifest,
	.manifest_contents = fc_bbsink_copystream_manifest_contents,
	.end_manifest = fc_bbsink_copystream_end_manifest,
	.end_backup = fc_bbsink_copystream_end_backup,
	.cleanup = fc_bbsink_copystream_cleanup
};

/*
 * 创建新的'copystream' bbsink。
 */
bbsink * bbsink_copystream_new(bool fc_send_to_client)
{
	bbsink_copystream *fc_sink = palloc0(sizeof(bbsink_copystream));

	*((const bbsink_ops **) &fc_sink->base.bbs_ops) = &bbsink_copystream_ops;
	fc_sink->send_to_client = fc_send_to_client;

	/* 设置定期进度报告。 */
	fc_sink->last_progress_report_time = GetCurrentTimestamp();
	fc_sink->bytes_done_at_last_time_check = UINT64CONST(0);

	return &fc_sink->base;
}

/*
 * 发送备份开始的线协议消息。
 */
static void fc_bbsink_copystream_begin_backup(bbsink *fc_sink)
{
	bbsink_copystream *fc_mysink = (bbsink_copystream *) fc_sink;
	bbsink_state *fc_state = fc_sink->bbs_state;
	char	   *fc_buf;

	/*
	 * 初始化缓冲区。我们最终希望通过CopyData消息发送档案和清单数据，其中每条消息的有效载荷部分以类型字节开头。然而，basebackup.c期望缓冲区是对齐的，因此我们不能仅为类型字节分配一个额外的字节。相反，分配足够的额外字节，以便我们向调用者揭示的缓冲区部分可以对齐，同时留下将类型字节提前插入的空间。这将允许我们通过单次调用pq_putmessage发送数据，而不需要任何额外的复制。
	 */
	fc_buf = palloc(fc_mysink->base.bbs_buffer_length + MAXIMUM_ALIGNOF);
	fc_mysink->msgbuffer = fc_buf + (MAXIMUM_ALIGNOF - 1);
	fc_mysink->base.bbs_buffer = fc_buf + MAXIMUM_ALIGNOF;
	fc_mysink->msgbuffer[0] = 'd'; /* 档案或清单数据 */

	/* 告诉客户端备份开始位置。 */
	fc_SendXlogRecPtrResult(fc_state->startptr, fc_state->starttli);

	/* 向客户端发送表空间列表。 */
	fc_SendTablespaceList(fc_state->tablespaces);

	/* 发送CommandComplete消息 */
	pq_puttextmessage('C', "SELECT");

	/* 开始COPY流。这将用于所有档案 + 清单。 */
	fc_SendCopyOutResponse();
}

/*
 * 发送CopyData消息，宣布新的档案开始。
 */
static void fc_bbsink_copystream_begin_archive(bbsink *fc_sink, const char *fc_archive_name)
{
	bbsink_state *fc_state = fc_sink->bbs_state;
	tablespaceinfo *fc_ti;
	StringInfoData fc_buf;

	fc_ti = list_nth(fc_state->tablespaces, fc_state->tablespace_num);
	pq_beginmessage(&fc_buf, 'd'); /* CopyData */
	pq_sendbyte(&fc_buf, 'n');		/* 新档案 */
	pq_sendstring(&fc_buf, fc_archive_name);
	pq_sendstring(&fc_buf, fc_ti->path == NULL ? "" : fc_ti->path);
	pq_endmessage(&fc_buf);
}

/*
 * 发送包含档案内容块的CopyData消息。
 */
static void fc_bbsink_copystream_archive_contents(bbsink *fc_sink, size_t fc_len)
{
	bbsink_copystream *fc_mysink = (bbsink_copystream *) fc_sink;
	bbsink_state *fc_state = fc_mysink->base.bbs_state;
	StringInfoData fc_buf;
	uint64		fc_targetbytes;

	/* 如果合适，将档案内容发送给客户端。 */
	if (fc_mysink->send_to_client)
	{
		/* 加一，因为我们也发送了一个领先的类型字节。 */
		pq_putmessage('d', fc_mysink->msgbuffer, fc_len + 1);
	}

	/* 考虑是否向客户端发送进度报告。 */
	fc_targetbytes = fc_mysink->bytes_done_at_last_time_check
		+ PROGRESS_REPORT_BYTE_INTERVAL;
	if (fc_targetbytes <= fc_state->bytes_done)
	{
		TimestampTz fc_now = GetCurrentTimestamp();
		long		fc_ms;

		/*
		 * 好吧，我们发送了相当数量的字节，所以检查系统时间以查看我们是否应该发送进度报告。
		 */
		fc_mysink->bytes_done_at_last_time_check = fc_state->bytes_done;
		fc_ms = TimestampDifferenceMilliseconds(fc_mysink->last_progress_report_time,
											 fc_now);

		/*
		 * 如果足够的时间已经过去，则发送进度报告。如果系统时钟向后设置，也要发送一个，以便这种情况不会抑制进一步的进度消息。
		 */
		if (fc_ms < 0 || fc_ms >= PROGRESS_REPORT_MILLISECOND_THRESHOLD)
		{
			fc_mysink->last_progress_report_time = fc_now;

			pq_beginmessage(&fc_buf, 'd'); /* CopyData */
			pq_sendbyte(&fc_buf, 'p'); /* 进度报告 */
			pq_sendint64(&fc_buf, fc_state->bytes_done);
			pq_endmessage(&fc_buf);
			pq_flush_if_writable();
		}
	}
}

/*
 * 我们不需要明确地信号归档的结束；客户端
 * 会在我们开始下一个时自行判断是否到达结束，
 * 或开始清单，或结束COPY流。然而，这似乎是
 * 强制输出进度报告的好时机。一个原因是如果
 * 这是最后一个归档，如果我们现在不强制
 * 进度报告，客户端将永远不会被告知我们发送了所有字节。
 */
static void fc_bbsink_copystream_end_archive(bbsink *fc_sink)
{
	bbsink_copystream *fc_mysink = (bbsink_copystream *) fc_sink;
	bbsink_state *fc_state = fc_mysink->base.bbs_state;
	StringInfoData fc_buf;

	fc_mysink->bytes_done_at_last_time_check = fc_state->bytes_done;
	fc_mysink->last_progress_report_time = GetCurrentTimestamp();
	pq_beginmessage(&fc_buf, 'd'); /* CopyData */
	pq_sendbyte(&fc_buf, 'p');		/* 进度报告 */
	pq_sendint64(&fc_buf, fc_state->bytes_done);
	pq_endmessage(&fc_buf);
	pq_flush_if_writable();
}

/*
 * 发送一个CopyData消息，宣布备份清单的开始。
 */
static void fc_bbsink_copystream_begin_manifest(bbsink *fc_sink)
{
	StringInfoData fc_buf;

	pq_beginmessage(&fc_buf, 'd'); /* CopyData */
	pq_sendbyte(&fc_buf, 'm');		/* 清单 */
	pq_endmessage(&fc_buf);
}

/*
 * 每一块清单数据使用CopyData消息发送。
 */
static void fc_bbsink_copystream_manifest_contents(bbsink *fc_sink, size_t fc_len)
{
	bbsink_copystream *fc_mysink = (bbsink_copystream *) fc_sink;

	if (fc_mysink->send_to_client)
	{
		/* 加一，因为我们也发送了一个领先的类型字节。 */
		pq_putmessage('d', fc_mysink->msgbuffer, fc_len + 1);
	}
}

/*
 * 我们不需要备份清单的显式终止符。
 */
static void fc_bbsink_copystream_end_manifest(bbsink *fc_sink)
{
	/* 什么也不做。 */
}

/*
 * 发送备份结束的线协议消息。
 */
static void fc_bbsink_copystream_end_backup(bbsink *fc_sink, XLogRecPtr fc_endptr,
							 TimeLineID fc_endtli)
{
	fc_SendCopyDone();
	fc_SendXlogRecPtrResult(fc_endptr, fc_endtli);
}

/*
 * 清理。
 */
static void fc_bbsink_copystream_cleanup(bbsink *fc_sink)
{
	/* 没有什么可做的。 */
}

/*
 * 发送一个CopyOutResponse消息。
 */
static void fc_SendCopyOutResponse(void)
{
	StringInfoData fc_buf;

	pq_beginmessage(&fc_buf, 'H');
	pq_sendbyte(&fc_buf, 0);		/* 整体格式 */
	pq_sendint16(&fc_buf, 0);		/* natts */
	pq_endmessage(&fc_buf);
}

/*
 * 发送一个CopyDone消息。
 */
static void fc_SendCopyDone(void)
{
	pq_putemptymessage('c');
}

/*
 * 发送一个包含单个
 * XLogRecPtr记录（以文本格式）的结果集
 */
static void fc_SendXlogRecPtrResult(XLogRecPtr fc_ptr, TimeLineID fc_tli)
{
	StringInfoData fc_buf;
	char		fc_str[MAXFNAMELEN];
	Size		fc_len;

	pq_beginmessage(&fc_buf, 'T'); /* RowDescription */
	pq_sendint16(&fc_buf, 2);		/* 2个字段 */

	/* 字段头 */
	pq_sendstring(&fc_buf, "recptr");
	pq_sendint32(&fc_buf, 0);		/* 表 oid */
	pq_sendint16(&fc_buf, 0);		/* attnum */
	pq_sendint32(&fc_buf, TEXTOID);	/* 类型 oid */
	pq_sendint16(&fc_buf, -1);
	pq_sendint32(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);

	pq_sendstring(&fc_buf, "tli");
	pq_sendint32(&fc_buf, 0);		/* 表 oid */
	pq_sendint16(&fc_buf, 0);		/* attnum */

	/*
	 * int8在这里可能看起来是一个令人惊讶的数据类型，
	 * 但理论上int4对于这个来说不够宽，因为TimeLineID是无符号的。
	 */
	pq_sendint32(&fc_buf, INT8OID);	/* 类型 oid */
	pq_sendint16(&fc_buf, -1);
	pq_sendint32(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);
	pq_endmessage(&fc_buf);

	/* 数据行 */
	pq_beginmessage(&fc_buf, 'D');
	pq_sendint16(&fc_buf, 2);		/* 列数 */

	fc_len = snprintf(fc_str, sizeof(fc_str),
				   "%X/%X", LSN_FORMAT_ARGS(fc_ptr));
	pq_sendint32(&fc_buf, fc_len);
	pq_sendbytes(&fc_buf, fc_str, fc_len);

	fc_len = snprintf(fc_str, sizeof(fc_str), "%u", fc_tli);
	pq_sendint32(&fc_buf, fc_len);
	pq_sendbytes(&fc_buf, fc_str, fc_len);

	pq_endmessage(&fc_buf);

	/* 发送CommandComplete消息 */
	pq_puttextmessage('C', "SELECT");
}

/*
 * 通过libpq发送一个结果集，描述表空间列表。
 */
static void fc_SendTablespaceList(List *fc_tablespaces)
{
	StringInfoData fc_buf;
	ListCell   *fc_lc;

	/* 构造并发送目录信息 */
	pq_beginmessage(&fc_buf, 'T'); /* RowDescription */
	pq_sendint16(&fc_buf, 3);		/* 3个字段 */

	/* 第一个字段 - spcoid */
	pq_sendstring(&fc_buf, "spcoid");
	pq_sendint32(&fc_buf, 0);		/* 表 oid */
	pq_sendint16(&fc_buf, 0);		/* attnum */
	pq_sendint32(&fc_buf, OIDOID); /* 类型 oid */
	pq_sendint16(&fc_buf, 4);		/* typlen */
	pq_sendint32(&fc_buf, 0);		/* typmod */
	pq_sendint16(&fc_buf, 0);		/* 格式代码 */

	/* 第二个字段 - spclocation */
	pq_sendstring(&fc_buf, "spclocation");
	pq_sendint32(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);
	pq_sendint32(&fc_buf, TEXTOID);
	pq_sendint16(&fc_buf, -1);
	pq_sendint32(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);

	/* 第三个字段 - 大小 */
	pq_sendstring(&fc_buf, "size");
	pq_sendint32(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);
	pq_sendint32(&fc_buf, INT8OID);
	pq_sendint16(&fc_buf, 8);
	pq_sendint32(&fc_buf, 0);
	pq_sendint16(&fc_buf, 0);
	pq_endmessage(&fc_buf);

	foreach(fc_lc, fc_tablespaces)
	{
		tablespaceinfo *fc_ti = lfirst(fc_lc);

		/* 发送一个数据行消息 */
		pq_beginmessage(&fc_buf, 'D');
		pq_sendint16(&fc_buf, 3);	/* 列数 */
		if (fc_ti->path == NULL)
		{
			pq_sendint32(&fc_buf, -1); /* 长度 = -1 ==> NULL */
			pq_sendint32(&fc_buf, -1);
		}
		else
		{
			Size		fc_len;

			fc_len = strlen(fc_ti->oid);
			pq_sendint32(&fc_buf, fc_len);
			pq_sendbytes(&fc_buf, fc_ti->oid, fc_len);

			fc_len = strlen(fc_ti->path);
			pq_sendint32(&fc_buf, fc_len);
			pq_sendbytes(&fc_buf, fc_ti->path, fc_len);
		}
		if (fc_ti->size >= 0)
			fc_send_int8_string(&fc_buf, fc_ti->size / 1024);
		else
			pq_sendint32(&fc_buf, -1); /* NULL */

		pq_endmessage(&fc_buf);
	}
}

/*
 * 通过线协议发送一个64位整数作为字符串。
 */
static void fc_send_int8_string(StringInfoData *fc_buf, int64 fc_intval)
{
	char		fc_is[32];

	sprintf(fc_is, INT64_FORMAT, fc_intval);
	pq_sendint32(fc_buf, strlen(fc_is));
	pq_sendbytes(fc_buf, fc_is, strlen(fc_is));
}
